package com.example.concurrency.business.device;

import com.example.concurrency.business.device.dto.DeviceDTO;
import com.example.concurrency.business.device.dto.DeviceImportResult;
import com.example.concurrency.business.device.enums.DeviceImportResultEnum;
import com.example.concurrency.business.device.impl.DeviceImportServiceImpl;
import com.example.concurrency.business.device.task.DeviceImportTask;
import com.google.common.collect.ImmutableList;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <p>
 *
 * </p>
 *
 * @author Ashe
 * @version V1.2.0
 * @date 2019年08月21日 19:19
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify By: {修改人} 2019年08月21日
 * @modify reason: {方法名}:{原因}
 * ...
 */
public class BatchResourceImportApplication {

    //任务线程【采用FutureTask便于直接获取线程返回结果】
    private static BlockingQueue<FutureTask<DeviceImportResult>> queue = new ArrayBlockingQueue<>(1000);
    //线程缓存等待队列
    private static BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(300);
    //线程池定义
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 8, 300, TimeUnit.MILLISECONDS, workQueue);
    //单个设备导入结果默认成功的集合，用于判断后更新进度。
    private static Collection<DeviceImportResultEnum> END_RESULT = ImmutableList.of(
            DeviceImportResultEnum.DEVICE_IMPORT_RESULT_ENUM_IMPORT_FAIL,
            DeviceImportResultEnum.DEVICE_IMPORT_RESULT_ENUM_IMPORT_SUCCESS,
            DeviceImportResultEnum.DEVICE_IMPORT_RESULT_ENUM_VALIDATE_FAIL);
    private static final int IMPORT_BATCH_TOTAL = 1000;

    /**
     * 多线程消费模型 导入设备
     *
     * @param args
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        IDeviceImportService deviceImportService = new DeviceImportServiceImpl();
        int i = 0;
        while (i++ < IMPORT_BATCH_TOTAL) {
            try {
                queue.put(new FutureTask<>(new DeviceImportTask(deviceImportService,
                        DeviceDTO.builder()
                                .deviceName("device-name-" + i)
                                .rowIndex(i)
                                .build())));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        /**
         * 结果集合 用来计算进度， 通过 ReentrantLock 完成 ArrayList 的并发线程访问
         * TODO 待优化
         */
        List<DeviceImportResult> resultList = new ArrayList<>();
        Lock lock = new ReentrantLock();
        threadPoolExecutor.execute(new Consumer(resultList, lock));
        threadPoolExecutor.execute(new Consumer(resultList, lock));
        Thread.sleep(3000);
        while (resultList.size() <= IMPORT_BATCH_TOTAL) {
            System.out.println("importing....");
            Thread.sleep(1000);
            lock.lock();
            long count = resultList.stream().filter(p -> END_RESULT.contains(p.getResultEnum())).count();
            System.out.println(String.format("success process: %8.2f",
                    (Double.valueOf(count) / Double.valueOf(IMPORT_BATCH_TOTAL)) * 100));
            System.out.println(String.format("total process: %8.2f",
                    (Double.valueOf(resultList.size()) / Double.valueOf(IMPORT_BATCH_TOTAL)) * 100));
            lock.unlock();

        }

    }

    /**
     * 消费者
     */
    static class Consumer implements Runnable {

        /**
         * 结果集合（非线程安全，需要加锁，进行部分原子操作）
         */
        List<DeviceImportResult> resultList;

        private Lock lock;

        public Consumer(List<DeviceImportResult> resultList, Lock lock) {
            this.resultList = resultList;
            this.lock = lock;
        }

        /**
         * When an object implementing interface <code>Runnable</code> is used
         * to create a thread, starting the thread causes the object's
         * <code>run</code> method to be called in that separately executing
         * thread.
         * <p>
         * The general contract of the method <code>run</code> is that it may
         * take any action whatsoever.
         *
         * @see Thread#run()
         */
        @Override
        public void run() {
            while (!queue.isEmpty()) {
                try {
                    FutureTask<DeviceImportResult> task = queue.take();
                    threadPoolExecutor.execute(task);
                    Thread.sleep(1000);
                    lock.lock();
                    resultList.add(task.get());
                    lock.unlock();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

